package com.cloudlead.common.worker.impl;

import com.cloudlead.common.lang.CacheQueue;
import com.cloudlead.common.lang.Messages;
import com.cloudlead.common.lang.api.*;
import com.cloudlead.common.lang.api.ActionResponse.Level;
import com.cloudlead.common.lang.exception.BadRequestException;
import com.cloudlead.common.lang.exception.ServiceUnavailableException;
import com.cloudlead.common.lang.serialize.JSON;
import com.cloudlead.common.lang.utils.ExceptionHelper;
import com.cloudlead.common.lang.utils.ObjectId;
import com.cloudlead.common.worker.*;
import com.cloudlead.common.worker.action.Action;
import com.cloudlead.common.worker.action.ActionNotFoundException;
import com.cloudlead.common.worker.desc.ActionDesc;
import com.cloudlead.common.worker.desc.datatype.FormData;
import com.cloudlead.common.worker.desc.datatype.MapData;
import com.cloudlead.common.worker.desc.datatype.TypeDesc;
import com.cloudlead.utils.async.AsyncTask;
import com.cloudlead.utils.async.AsyncTaskHandle;
import com.cloudlead.utils.async.AsyncTaskService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.ResponseBuilder;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Service
public class ApiServiceImpl implements ApiService {
    private final static Logger logger = LoggerFactory
            .getLogger(ApiService.class);
    private static final int DEFAULT_LIVE_TIME_IN_SECONDS = 60 * 60;

    @Resource
    ApiRegistry apiRegistry;

    @Resource
    Interceptor[] interceptors;

    @Resource
    private AsyncTaskService asyncTaskService;

    private CacheQueue<String> executingRequests;

    private Map<String, ActionCancelHandle> actionExecuteQueue;

    @PostConstruct
    public void start() {
        executingRequests = new CacheQueue<>();
        actionExecuteQueue = new ConcurrentHashMap<>();
    }

    @PreDestroy
    public void stop() {
    }

    @Override
    public void execute(String actionName, String requestString,
                        Map<String, String> requestHeaders, AsyncResponse asyncResponse) {

        SimpleActionRequest request;
        try {
            ActionDesc apiInfo = apiRegistry.getAllApiDesc().get(actionName);
            if (null == apiInfo) {
                throw new ActionNotFoundException(Messages.getMessage("worker.action.NotFound", actionName));
            }
            request = deserializeRequest(apiInfo, requestString);
        } catch (Exception e) {
            e.printStackTrace();
            ErrorActionResponse errorActionResponse = ExceptionHelper.toResponse(new BadRequestException(e.getMessage(), e));
            asyncResponse.resume(errorActionResponse);
            return;
        }
        request.setRequestHeaders(requestHeaders);

        Action<ActionRequest, ActionResponse> action = apiRegistry
                .get(actionName);

        if (null == request.getRequestId()
                || request.getRequestId().length() == 0) {
            request.setRequestId(ObjectId.get());
        }


        // 还在执行的请求
        if (actionExecuteQueue.containsKey(request.getRequestId())) {

            if (request.isCancel()) {// 取消请求
                actionExecuteQueue.get(request.getRequestId()).cancel();
                EmptyActionResponse actionResponse = new EmptyActionResponse();
                actionResponse.setRequestId(request.getRequestId());
                asyncResponse.resume(Response.status(actionResponse.getCode())
                        .entity(actionResponse).build());
            } else {// 说明请求还在执行
                AsyncActionResponse asyncActionResponse = new AsyncActionResponse();
                asyncActionResponse.setRequestId(request.getRequestId());
                asyncResponse.resume(Response
                        .status(asyncActionResponse.getCode())
                        .entity(asyncActionResponse).build());
            }

        } else {
            //表单重复提交
            if (executingRequests.contains(request.getRequestId())) {
                RepeatActionResponse repeatActionResponse = new RepeatActionResponse();
                repeatActionResponse.setRequestId(request.getRequestId());
                asyncResponse.resume(Response
                        .status(repeatActionResponse.getCode())
                        .entity(repeatActionResponse).build());
            } else {
                asyncExecutionAction(action, actionName, request, asyncResponse);
            }
        }

    }

    private SimpleActionRequest deserializeRequest(ActionDesc apiInfo, String requestString) {
        SimpleActionRequest simpleActionRequest = JSON.toBean(requestString, SimpleActionRequest.class);
        Map<String, Object> requestContent = simpleActionRequest.getContent();
        Map<String, Object> params = new HashMap<>();
        MapData requestMap = apiInfo.getRequest();
        Map<String, FormData> mapping = ((MapData) requestMap.getMapping().get("content")).getMapping();
        for (Entry<String, FormData> entry : mapping.entrySet()) {
            String paramName = entry.getKey();
            FormData paramType = entry.getValue();
            Class<?> rawType = paramType.getJavaType().getRawType();
            Class contentType = null;
            List<TypeDesc> elementType = paramType.getJavaType().getElementType();
            if (null != elementType && !elementType.isEmpty()) {
                TypeDesc typeDesc = elementType.get(0);
                contentType = typeDesc.getRawType();
            }
            if (null != requestContent) {
                Object content = JSON.toBean(JSON.toString(requestContent.get(paramName)), rawType, contentType);
                params.put(paramName, content);
            } else {
                params.put(paramName, null);
            }
        }
        simpleActionRequest.setContent(params);
        return simpleActionRequest;
    }

    /**
     * 执行动作
     *
     * @param action
     * @param actionName
     * @param request
     * @param asyncResponse
     */
    private void asyncExecutionAction(
            final Action<ActionRequest, ActionResponse> action,
            final String actionName, final ActionRequest request,
            final AsyncResponse asyncResponse) {

        // 处理超时
        if (0 < request.getTimeout()) {
            asyncResponse.setTimeoutHandler(asyncResponse1 -> {
                SimpleActionResponse<String> response = new SimpleActionResponse<>(
                        false, Level.warn, Messages.getMessage("worker.action.execute.timeout"),
                        ResponseCode.REQUEST_TIMEOUT.code(),
                        ResponseCode.REQUEST_TIMEOUT.reason());
                response.setRequestId(request.getRequestId());

                asyncResponse1.resume(Response.ok()
                        .entity(response).build());
            });
            asyncResponse.setTimeout(request.getTimeout(), TimeUnit.SECONDS);
        }

        // 异步执行任务
        final AsyncTaskHandle<ActionResponse> taskHandle = asyncTaskService
                .addTask(new AsyncTask<ActionResponse>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public ActionResponse call() throws Exception {
                        executingRequests.put(request.getRequestId(), DEFAULT_LIVE_TIME_IN_SECONDS);

                        ActionContext actionConfig = new ActionContext();
                        actionConfig.setRequest(request);
                        actionConfig.setAction(action);
                        actionConfig.setActionName(actionName);

                        // 执行过滤器链
                        InterceptorChainImpl interceptorChain = new InterceptorChainImpl(
                                interceptors);
                        interceptorChain.process(actionConfig);

                        ActionResponse response = actionConfig.getResponse();
                        if (null == response) {
                            response = new EmptyActionResponse();
                        }
                        ((AbstractActionResponse) response)
                                .setRequestId(request.getRequestId());
                        actionExecuteQueue.remove(request.getRequestId());

                        ResponseBuilder resp = Response.ok();
                        if (!response.getHeaders().isEmpty()) {
                            Set<Entry<String, String>> entrySet = response
                                    .getHeaders().entrySet();
                            for (Entry<String, String> entry : entrySet) {
                                resp.header(entry.getKey(), entry.getValue());
                            }
                        }
                        asyncResponse.resume(resp.entity(response).build());
                        return response;
                    }

                });

        // 添加一个Action取消操作句柄
        actionExecuteQueue.put(request.getRequestId(), () -> {
            taskHandle.cancel(true);
            ErrorActionResponse response = ExceptionHelper
                    .toResponse(new ServiceUnavailableException(
                            Messages.getMessage("worker.action.execute.cancel")));
            response.setRequestId(request.getRequestId());

            asyncResponse.resume(Response.ok().entity(response)
                    .build());
            logger.debug("request [" + request.getRequestId()
                    + "] cancel success.");

        });

    }

}
